package com.rocketmq.consumer.utils;

import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicLong;

@Slf4j
public class StatisticCount {

	private static final ScheduledExecutorService LOG_SERVICE;

	static {
		LOG_SERVICE = Executors.newScheduledThreadPool(1, new ThreadFactory() {
			@Override
			public Thread newThread(Runnable r) {
				Thread thread = new Thread(r);
				thread.setDaemon(true);
				thread.setName("StatisticCountThread");
				return thread;
			}

		});
	}

	private static final ConcurrentMap<String, StatisticCount> CACHE = new ConcurrentHashMap<>();

	private String cName;
	private final AtomicLong COUNTER = new AtomicLong(0);
	private long COUNTER_OLD = 0;

	private Runnable logTask;

	private StatisticCount() {
	}

	/**
	 * @Description  
	 * @param counterName 被统计项目的名字
	 * @param logInterval 日志打印间隔，单位s	
	 * @return
	 */
	public static StatisticCount getCounter(String counterName, int logInterval) {
		if (logInterval <= 0) {
			logInterval = 60;
		}
		final int interval = logInterval;
		StatisticCount ret = CACHE.get(counterName);
		if (ret == null) {
			StatisticCount c = new StatisticCount();
			c.cName = counterName;
			c.logTask = new Runnable() {

				@Override
				public void run() {
					long count = c.COUNTER.get();
					log.info("concurrent for {}: {}/{}s", c.cName, count - c.COUNTER_OLD, interval);
					c.COUNTER_OLD = count;
				}

			};
			if (CACHE.putIfAbsent(counterName, c) == null) {
				LOG_SERVICE.scheduleAtFixedRate(c.logTask, interval, interval, TimeUnit.SECONDS);
			}
		}
		return CACHE.get(counterName);
	}

	public static StatisticCount getCounter(String counterName) {
		return getCounter(counterName, 60);
	}

	public void addCount(long num) {
		COUNTER.addAndGet(num);
	}

}
